package com.salmon.study.actors;


import akka.actor.ActorRef;
import akka.actor.UntypedAbstractActor;
import com.salmon.study.message.MapData;
import com.salmon.study.message.ReduceData;
import com.salmon.study.message.WordCount;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;

@Slf4j
public class ReduceActor extends UntypedAbstractActor {

    private ActorRef aggregateActor;

    public ReduceActor(ActorRef aggregateActor) {
        this.aggregateActor = aggregateActor;
    }

    @Override
    public void onReceive(Object message) throws Throwable {
        log.info("{}", message);
        if(message instanceof MapData) {
            MapData mapData = (MapData) message;
            ReduceData reduceData = reduce(mapData.getDataList());
            this.aggregateActor.tell(reduceData, ActorRef.noSender());
        }  else {
            unhandled(message);
        }
    }

    private ReduceData reduce(List<WordCount> dataList) {
        HashMap<String, Integer> reduceMap = new HashMap<>();
        for (WordCount wordCount : dataList) {
            if(reduceMap.containsKey(wordCount.getWord())) {
                int value = reduceMap.get(wordCount.getWord());
                reduceMap.put(wordCount.getWord(),++value);
            } else {
                reduceMap.put(wordCount.getWord(), 1);
            }
        }
        return new ReduceData(reduceMap);
    }
}


